# -*- coding: utf-8 -*-
'''
@author: olivia.dou
Created on: 2019/3/28 13:30
desc: 验证kafka消息队列中数据即时被消费入库到mysql
'''

import time,datetime,json,unittest,logging
from kafka import KafkaConsumer #kafka-python
from util.db_util import execute_query

class TestKafkaMessage(unittest.TestCase):


    def setUp(self):
        self.bootstrap_servers =  ["192.168.250.203:9092","192.168.250.204:9092","192.168.250.205:9092"]
        logger = logging.getLogger("kafka")
        logger.setLevel(logging.WARN)

    @unittest.skip("")
    def test_news_topic(self):

        # ch = logging.StreamHandler()
        # formatter = logging.Formatter('%(name)-12s: %(levelname)-8s %(message)s')
        # ch.setFormatter(formatter)
        # logging.getLogger('').addHandler(ch)

        news_consumer = KafkaConsumer(bootstrap_servers = self.bootstrap_servers)
        news_consumer.subscribe(topics=('qa_news_topic',))

        # 持续获取kafka数据
        #news_consumer = KafkaConsumer('qa_news_topic',bootstrap_servers = bootstrap_servers)

        # for msg in news_consumer:
        #      #print(str(msg.value,'utf-8'))
        #     print(msg)

        #report_consumer = KafkaConsumer(bootstrap_servers = self.bootstrap_servers)
        #report_consumer.subscribe(topics=('knews_report_qa',))
        #msg = report_consumer.poll(timeout_ms=5000)  # 从kafka获取消息
        #print(msg)

        index = 0
        titles = []
        while len(titles)<10:
            logging.info('--------poll index is %s----------' % index)
            logging.info(datetime.datetime.now())
            msg = news_consumer.poll(timeout_ms=1000,max_records=10)  # 从kafka获取消息
            if msg != {}:
                for value in list(msg.values()):
                    logging.info("new value:")
                    vstr = str(value[0].value,'utf-8')
                    logging.info(vstr)
                    #print(json.loads(vstr)[0]['title'])
                    titles.append(json.loads(vstr)[0]['title'])

            time.sleep(1)
            index += 1

        logging.info(titles)

        time.sleep(60)

        for title in titles:
            res = execute_query("select * from nlp_news_feed where title = '%s'"%title)
            assert res!=[]


    def test_announcement_topic(self):

        push_types = ['100603','100401','160301','160302','160303','160304','160305','160306','160307']

        news_consumer = KafkaConsumer(bootstrap_servers = self.bootstrap_servers)
        news_consumer.subscribe(topics=('announcement_to_knews_qa',))

        index = 0
        ref_ids_push = []
        ref_ids_no_push = []
        logging.info("polling announcement")
        while len(ref_ids_push)<10 or ref_ids_no_push<10:
            logging.info('--------poll index is %s----------' % index)
            logging.info(datetime.datetime.now())
            msg = news_consumer.poll(timeout_ms=1000,max_records=10)  # 从kafka获取消息
            if msg != {}:
                for value in list(msg.values()):
                    logging.info("new message:")
                    vstr = str(value[0].value,'utf-8')
                    logging.info(vstr)
                    #print(json.loads(vstr)[0]['title'])
                    data = json.loads(vstr)[0]
                    if data['typ'] in push_types or data['algo_typ'] in push_types or data['manual_typ'] in push_types:
                        ref_ids_push.append(data['fn'])
                    else:
                        ref_ids_no_push.append(data['fn'])

            time.sleep(1)
            index += 1

        logging.info(titles)

        time.sleep(60)

        for id in ref_ids_push:
            res = execute_query("select * from nlp_news_feed where source_ref_id = '%s'"%id)
            assert res!=[]
        for id in ref_ids_no_push:
            res = execute_query("select * from nlp_news_feed where source_ref_id = '%s'"%id)
            assert res==[]

